package testEsClient;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.elasticsearch.index.query.QueryBuilders.rangeQuery;
import static org.elasticsearch.index.query.QueryBuilders.termsQuery;

import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.TimeZone;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.EventBuilder;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AggregationBuilders;

import com.eluup.flume.sink.elasticsearch.ElasticSearchScriptSerializer;
import com.eluup.flume.sink.elasticsearch.TimeBasedIndexNameBuilder;
import com.eluup.flume.sink.elasticsearch.client.ElasticSearchTransportScriptClient;
import com.eluup.flume.sink.elasticsearch.client.PreBuiltTransportClient;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.LongSerializationPolicy;
import com.google.gson.reflect.TypeToken;

public class EsClient {
	private static final SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");

	static String[] ESLIST = new String[] { "10.207.6.235"};
	static String[] ACTNAMELIST = new String[] { "fangwenlujing", "fasongxiaoxiKey", "qiangxiangeitafaxiaoxiKey", "shouyekehuxiaoxikey","xiaoxiqianketuijiankey", "fangniujiaxiaomishutuwenxiaoxikey"};
	
	private static PreBuiltTransportClient transportClient;
	

	private static void testEs() throws Exception {

		Context context = new Context();
		context.put("indexName", "esf_bi_user_feature_test");
		String header = "{\"topic\":\"esf_bi_user_feature\",\"key\":\"45209c8849bd5e94f984446d939e5db6\",\"timestamp\":\"1513039846204\"}";
		Gson gson = new GsonBuilder().setLongSerializationPolicy(LongSerializationPolicy.STRING).create();
		Type type = new TypeToken<Map<String, Object>>() {}.getType();
		Map<String, String> headers = gson.fromJson(new String(header.getBytes(), Charset.defaultCharset()), type);
		headers.put("indexName", "esf_bi_user_feature_test");
		String source = "{\"Update\":\"access_user_uid^lejuglobal^city\",\"house_geo\":[114.205196,30.51256],\"avgprice\":\"14447\",\"city\":\"wh\",\"district\":\"4625\",\"Append\":\"companyid^agentid^houseid^sinaid^block^district^buildingarea^price^house_geo^avgprice\",\"block\":\"4628\",\"sinaid\":\"281766\",\"lejuglobal\":\"123123\",\"ctime\":\"1512107649000\"}";
		Event event = EventBuilder.withBody(source.getBytes(), headers);
		ElasticSearchScriptSerializer serializer = new ElasticSearchScriptSerializer();
		ElasticSearchTransportScriptClient client = new ElasticSearchTransportScriptClient(ESLIST,"club", serializer);
		TimeBasedIndexNameBuilder indexNameBuilder = new TimeBasedIndexNameBuilder();
		indexNameBuilder.configure(context);
		client.addEvent(event, indexNameBuilder, "access", 5);
		client.execute();
		System.exit(0);
	 }
	public static void main(String[] args) throws Exception {
		
		testEs();
		 
		 
		
		
		
		
		String fromtoindex = "会话ID,消息快速查询字段";
		
		String end = "2017-03-21T09:29:09";
        Date d = formatter.parse(end);
        long start_l = d.getTime() - 3600000;
        String start = formatter.format(new Date(start_l));
        QueryBuilder qb = rangeQuery("@message.ctime")
        	    .from(start)
        	    .to(end)
        	    .includeLower(true)
        	    .includeUpper(false);
        QueryBuilder qb1 = termsQuery("@message.fromtoindex", fromtoindex);  
        SearchRequestBuilder sb = transportClient.prepareSearch("esf_im_res*");
        sb.addAggregation(AggregationBuilders.sum("isNewTopic").field("@message.isNewTopic"));
        sb.addAggregation(AggregationBuilders.sum("isAbleTopic").field("@message.isAbleTopic"));
        sb.addAggregation(AggregationBuilders.cardinality("fromNumber").field("@message.from_uid"));
        sb.setQuery(qb).setQuery(qb1).setSize(0);
        
        System.out.println(sb);
        SearchResponse searchresponse = sb.get();
        
		long count = searchresponse.getHits().getTotalHits();
		
		System.out.println(count);
		
		Double isNewTopicRes = (Double) searchresponse.getAggregations().get("isNewTopic").getProperty("value");
		Double isAbleTopicRes = (Double) searchresponse.getAggregations().get("isAbleTopic").getProperty("value");
		Double fromNumber = (Double) searchresponse.getAggregations().get("fromNumber").getProperty("value");
		
		int isNewTopic = 0, isAbleTopic = 0;
		System.out.println(isNewTopic);
		System.out.println(isAbleTopic);
		if (count == 0) {
			isNewTopic = 1;
		} else if (count > 0 && isNewTopicRes < 1) {
			isNewTopic = 1;
		} else {
			isNewTopic = 0;
		}
		
		if (fromNumber > 1 && isAbleTopicRes < 1) {
			isAbleTopic = 1;
		} else {
			isAbleTopic = 0;
		}
		
		for ( SearchHit searchhit : searchresponse.getHits().getHits()) {
			System.out.println(searchhit.sourceAsString());
			System.out.println(searchhit.getSource().get("@message"));
			Map<String, String> res = (Map<String,String>) searchhit.getSource().get("@message");
//			System.out.println(res.get("from_uid"));
		}
		
		System.exit(0);
		
		try {
	        
			IndexResponse response = transportClient.prepareIndex("twitter123", "tweet", "1")
			        .setSource(jsonBuilder()
			                    .startObject()
			                        .field("user", "kimchy")
			                        .field("postDate", new Date())
			                        .field("message", "trying out Elasticsearch")
			                    .endObject()
			                  )
			        .get();
			System.out.println(response.getResult());
		} catch (Exception e) {
			System.out.println(e.getMessage());
		}
		
		try {
			
			
			System.out.println("=====================");
			System.out.println("=====================");
			System.out.println("=====================");
			System.out.println("=====================");
			
			
			BulkRequestBuilder bulkRequest = transportClient.prepareBulk();
			bulkRequest.add(transportClient.prepareIndex("esf_bi_access-123", "tweet", "2")
			        .setSource(jsonBuilder()
			                    .startObject()
			                        .field("user", "kimchy")
			                        .field("postDate", new Date())
			                        .field("message", "another post")
			                        .field("message2", "another post")
			                        .field("message3", "another post")
			                        .field("message4", "another post")
			                    .endObject()
			                  )
			        );
			System.out.println(bulkRequest.request().toString());
			BulkResponse bulkResponse = bulkRequest.get();
			for (Iterator<BulkItemResponse> iterator = bulkResponse.iterator(); iterator.hasNext();) {
		        BulkItemResponse r = iterator.next();
		        System.out.println(r.getResponse().getResult());
	        }
			System.out.println();
			if (bulkResponse.hasFailures()) {
			    // process failures by iterating through each bulk response item
				System.out.println(bulkResponse.toString());
				System.out.println(bulkResponse.buildFailureMessage());
			}
		} catch (Exception e) {
			System.out.println(e.getMessage());
		}
		
	}
}
